package net.qiqbframework.eventsourcing;

import lombok.extern.slf4j.Slf4j;
import net.qiqbframework.common.Assert;
import net.qiqbframework.eventhandling.DomainEventMessage;
import net.qiqbframework.eventhandling.EventMessage;
import net.qiqbframework.eventhandling.distributing.EventDistributor;
import net.qiqbframework.messaging.intercepting.MessageDispatchInterceptor;
import net.qiqbframework.messaging.monitoring.MessageMonitor;
import net.qiqbframework.messaging.monitoring.MonitorCallback;
import net.qiqbframework.messaging.monitoring.NoOpMessageMonitor;
import net.qiqbframework.messaging.task.CurrentMessageTask;
import net.qiqbframework.messaging.task.MessageTask;
import net.qiqbframework.messaging.task.TaskPhase;
import net.qiqbframework.persisthanding.DefaultEventBusSpanFactory;
import net.qiqbframework.persisthanding.EventBusSpanFactory;
import net.qiqbframework.serialization.Serializer;
import net.qiqbframework.tracing.NoOpSpanFactory;
import net.qiqbframework.tracing.Span;
import net.qiqbframework.tracing.SpanScope;

import javax.annotation.Nonnull;
import java.util.ArrayList;
import java.util.List;
import java.util.Set;
import java.util.concurrent.CopyOnWriteArraySet;
import java.util.function.BiFunction;
import java.util.function.Consumer;
import java.util.function.Supplier;
import java.util.stream.Collectors;

import static net.qiqbframework.common.BuilderUtils.assertNonNull;


/**
 *
 */
@Slf4j
public abstract class AbstractEventStorageEngine implements EventStorageEngine {

    private final EventBusSpanFactory spanFactory;

    private final MessageMonitor<? super EventMessage<?>> messageMonitor;


    private final Set<MessageDispatchInterceptor<? super EventMessage<?>>> dispatchInterceptors = new CopyOnWriteArraySet<>();

    private final String eventsKey = this + "_EVENTS";

    private final Serializer eventSerializer;

    private EventDistributor eventDistributor;

    public Serializer getEventSerializer() {
        return eventSerializer;
    }

    protected AbstractEventStorageEngine(Builder builder) {
        // builder.validate();
        this.spanFactory = builder.spanFactory;
        this.messageMonitor = builder.messageMonitor;
        this.eventDistributor = builder.eventDistributor;
        this.eventSerializer = builder.eventSerializer.get();
    }


    @Override
    public void appendEvents(@Nonnull List<? extends EventMessage<?>> events) {

        List<? extends EventMessage<?>> eventsWithContext = events
                .stream()
                .map(e -> spanFactory.createPublishEventSpan(e)
                        .runSupplier(() -> spanFactory.propagateContext(e)))
                .collect(Collectors.toList());
        List<MonitorCallback> ingested = eventsWithContext.stream()
                .map(messageMonitor::onMessageIngested)
                .toList();

        if (CurrentMessageTask.isStarted()) {
            MessageTask<?> unitOfWork = CurrentMessageTask.get();
            Assert.state(!unitOfWork.phase().isAfter(TaskPhase.PREPARE_COMMIT),
                    () -> "It is not allowed to publish events when the current Unit of Work has already been " +
                            "committed. Please start a new Unit of Work before publishing events.");
            Assert.state(!unitOfWork.root().phase().isAfter(TaskPhase.PREPARE_COMMIT),
                    () -> "It is not allowed to publish events when the root Unit of Work has already been " +
                            "committed.");

            unitOfWork.afterCommit(u -> ingested.forEach(MonitorCallback::reportSuccess));
            unitOfWork.onRollback(uow -> ingested.forEach(
                    message -> message.reportFailure(uow.getExecutionResult().getExceptionResult())
            ));

            final List<EventMessage<?>> eventMessages = eventsQueue(unitOfWork);
            eventMessages.addAll(eventsWithContext);

        } else {

            spanFactory.createCommitEventsSpan().run(() -> {
                try {
                    prepareCommit(intercept(eventsWithContext));
                    commit(eventsWithContext);
                    afterCommit(eventsWithContext);
                    ingested.forEach(MonitorCallback::reportSuccess);
                } catch (Exception e) {
                    ingested.forEach(m -> m.reportFailure(e));
                    throw e;
                }
            });
        }
    }

    private List<EventMessage<?>> eventsQueue(MessageTask<?> unitOfWork) {
        return unitOfWork.getOrComputeResource(eventsKey, r -> {
            Span commitSpan = spanFactory.createCommitEventsSpan();
            List<EventMessage<?>> eventQueue = new ArrayList<>();

            unitOfWork.onPrepareCommit(u -> {
                commitSpan.start();
                try (SpanScope unused = commitSpan.makeCurrent()) {
                    if (u.parent().isPresent() && !u.parent().get().phase().isAfter(TaskPhase.PREPARE_COMMIT)) {
                        eventsQueue(u.parent().get()).addAll(eventQueue);
                    } else {
                        int processedItems = eventQueue.size();
                        doWithEvents(this::prepareCommit, intercept(eventQueue));
                        // Make sure events published during publication prepare commit phase are also published
                        while (processedItems < eventQueue.size()) {
                            List<? extends EventMessage<?>> newMessages = intercept(eventQueue.subList(processedItems, eventQueue.size()));
                            processedItems = eventQueue.size();
                            doWithEvents(this::prepareCommit, newMessages);
                        }
                    }
                }
            });
            unitOfWork.onCommit(u -> {
                try (SpanScope unused = commitSpan.makeCurrent()) {
                    if (u.parent().isPresent() && !u.root().phase().isAfter(TaskPhase.COMMIT)) {
                        u.root().onCommit(w -> doWithEvents(this::commit, eventQueue));
                    } else {
                        doWithEvents(this::commit, eventQueue);
                    }
                }
            });
            unitOfWork.afterCommit(u -> {
                try (SpanScope unused = commitSpan.makeCurrent()) {
                    if (u.parent().isPresent() && !u.root().phase().isAfter(TaskPhase.AFTER_COMMIT)) {
                        log.info("appendEvents eventQueue root {}",eventQueue.size());
                       // u.root().afterCommit(w -> doWithEvents(this::afterCommit, eventQueue));
                    } else {
                        log.info("appendEvents eventQueue {}",eventQueue.size());
                        doWithEvents(this::afterCommit, eventQueue);
                    }
                }
            });
            unitOfWork.onCleanup(u -> {
                u.resources().remove(eventsKey);
                commitSpan.end();
            });
            return eventQueue;
        });
    }

    private void doWithEvents(Consumer<List<? extends EventMessage<?>>> eventsConsumer,
                              List<? extends EventMessage<?>> events) {
        eventsConsumer.accept(events);
    }

    /**
     * Check whether or not this is the first event, which means we tried to create an aggregate through the given
     * {@code failedEvent}.
     *
     * @param failedEvent the event to be checked
     * @return true in case of first event, false otherwise
     */
    private boolean isFirstDomainEvent(EventMessage<?> failedEvent) {
        if (failedEvent instanceof DomainEventMessage<?>) {
            return ((DomainEventMessage<?>) failedEvent).getSequenceNumber() == 0L;
        }
        return false;
    }

    protected List<? extends EventMessage<?>> intercept(List<? extends EventMessage<?>> events) {
        List<EventMessage<?>> preprocessedEvents = new ArrayList<>(events);
        for (MessageDispatchInterceptor<? super EventMessage<?>> preprocessor : dispatchInterceptors) {
            BiFunction<Integer, ? super EventMessage<?>, ? super EventMessage<?>> function =
                    preprocessor.handle(preprocessedEvents);
            for (int i = 0; i < preprocessedEvents.size(); i++) {
                preprocessedEvents.set(i, (EventMessage<?>) function.apply(i, preprocessedEvents.get(i)));
            }
        }
        return preprocessedEvents;
    }

    protected void prepareCommit(List<? extends EventMessage<?>> events) {

    }

    /**
     * Process given {@code events} while the Unit of Work root is being committed. The default implementation does
     * nothing.
     *
     * @param events Events to be published by this Event Bus
     */
    protected void commit(List<? extends EventMessage<?>> events) {
    }

    /**
     * Process given {@code events} after the Unit of Work has been committed. The default implementation does nothing.
     *
     * @param events Events to be published by this Event Bus
     */
    protected void afterCommit(List<? extends EventMessage<?>> events) {
        log.info("afterCommit {}",events.size());
        for (EventMessage<?> event : events) {
            eventDistributor.distributing(event);
        }

    }


    public abstract static class Builder {

        private MessageMonitor<? super EventMessage<?>> messageMonitor = NoOpMessageMonitor.INSTANCE;
        private EventBusSpanFactory spanFactory = DefaultEventBusSpanFactory
                .builder().spanFactory(NoOpSpanFactory.INSTANCE).build();
        private Supplier<Serializer> eventSerializer;
        private EventDistributor eventDistributor;

        public Builder messageMonitor(@Nonnull MessageMonitor<? super EventMessage<?>> messageMonitor) {
            assertNonNull(messageMonitor, "MessageMonitor may not be null");
            this.messageMonitor = messageMonitor;
            return this;
        }


        public Builder spanFactory(@Nonnull EventBusSpanFactory spanFactory) {
            assertNonNull(spanFactory, "SpanFactory may not be null");
            this.spanFactory = spanFactory;
            return this;
        }

        public Builder eventDistributor(EventDistributor eventDistributor) {
            assertNonNull(eventDistributor, "The event  Distributor may not be null");
            this.eventDistributor = eventDistributor;
            return this;
        }

        public Builder eventSerializer(Serializer eventSerializer) {
            assertNonNull(eventSerializer, "The event Serializer may not be null");
            this.eventSerializer = () -> eventSerializer;
            return this;
        }


    }
}
